package org.debezium.getInformation;

import io.debezium.data.Envelope;
import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.debezium.utils.CommonUtil;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import static io.debezium.data.Envelope.FieldName.*;
import static java.util.stream.Collectors.toMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class debeziumConnectorInformations {
    private Properties properties;

    public debeziumConnectorInformations() {
        Logger logger = LoggerFactory.getLogger("a");
        logger.debug("debug");
        logger.error("error");
        logger.info("info");
        CommonUtil commonUtil = new CommonUtil();
        properties = commonUtil.loadProperties("debezium-connector.properties");
        properties.put("database.history.instance.name", UUID.randomUUID().toString());
        createDebeziumEngine();
    }

    public void createDebeziumEngine() {
        DebeziumEngine<RecordChangeEvent<SourceRecord>> engine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(properties)
                .notifying(this::handlePayload)
                .build();
        engine.run();
    }

    public void handlePayload(List<RecordChangeEvent<SourceRecord>> recordChangeEvents, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> recordRecordCommitter) {
        recordChangeEvents.forEach(
                recordRecordChangeEvent -> {
                    SourceRecord sourceRecord = recordRecordChangeEvent.record();
//                    System.out.println("sourceRecord: "+sourceRecord);
                    Struct sourceRecordChangeValue = (Struct) sourceRecord.value();
                    Envelope.Operation operation = Envelope.Operation.forCode((String) sourceRecordChangeValue.get(OPERATION));
                    //如果是删除操作，获取操作前数据；否则获取操作后数据
                    String record = operation == Envelope.Operation.DELETE ? BEFORE : AFTER;
                    Struct struct = (Struct) sourceRecordChangeValue.get(record);
                    //封装变更数据为map
                    Map<String, Object> payload = struct.schema().fields().stream()
                            .map(Field::name)
                            .filter(fieldName -> struct.get(fieldName) != null)
                            .map(fieldName -> Pair.of(fieldName, struct.get(fieldName)))
                            .collect(toMap(Pair::getKey, Pair::getValue));
                    Struct dbTb = (Struct) sourceRecordChangeValue.get("source");
                    String db = dbTb.get("db").toString();
                    String tb = dbTb.get("table").toString();
                    System.out.println(db+"."+tb+"表，数据为: "+payload);
                    //TODO 自定义数据处理
                    //可根据需要过滤掉获取到的signal表数据


                    //提交offset
                    try {
                        recordRecordCommitter.markProcessed(recordRecordChangeEvent);
                        recordRecordCommitter.markBatchFinished();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

        );
    }
}
